Summary

MirrorSourceTask currently relies on the source consumer's
auto.offset.reset policy to recover from any OffsetOutOfRangeException.
In production that means the two most damaging real-world failure modes are
indistinguishable from a healthy stream:

1. Log truncation — the source log's start offset advances past the
   replication position (retention policy, kafka-delete-records.sh, etc.).
   With auto.offset.reset=earliest the consumer silently jumps forward; the
   target cluster ends up with a gap that no metric or log line reflects.
2. Topic reset — the source topic is deleted and recreated, e.g. during
   maintenance. The consumer's stored position is now beyond the new
   endOffset, so it either throws repeatedly or, again with earliest,
   silently jumps forward.

This change makes MirrorSourceTask recognise both situations and react
differently.

Changes

A new catch (OffsetOutOfRangeException e) branch in poll() dispatches to a
new helper handleOutOfRangeOffsets(...):

• If position > endOffset && beginningOffset == 0 for an out-of-range
  partition, the source topic looks newly recreated. The task logs a WARN
  with the timestamp and topic-partition, calls consumer.seek(tp, 0L), and
  resumes replication on the next poll.
• If position < beginningOffset, data we still needed to replicate has been
  purged. The task logs an ERROR describing the gap (positions and the
  number of records lost) and throws a ConnectException, surfacing the
  failure to Connect's task lifecycle (fail-fast) instead of papering over it.
• Any other shape of out-of-range is treated as unexpected and fails the
  task, so the implementation degrades safely rather than silently.

initializeConsumer(...) was also tightened: partitions with no committed
offset now seekToBeginning(...) explicitly. This makes the task behave
correctly when operators choose consumer.auto.offset.reset = none to opt
into the truncation detection above (otherwise a first-time partition would
throw NoOffsetForPartitionException on first poll).

Net change: one file, ~70 added / ~7 removed.

Why the predicate is safe

The beginningOffset == 0 condition in the reset branch is load-bearing:
ordinary retention-driven trimming always leaves beginningOffset > 0, so
retention can never be mistaken for a topic reset — it always falls into the
truncation branch. Conversely, a recreated topic has both
beginningOffset == 0 and an endOffset lower than what we previously
read, so the recreation case has a clean discriminator.

Testing

• Unit logic was exercised with a Docker Compose harness (two single-node
  KRaft clusters + this build of MM2 + a small producer). Three scenarios are
  run end-to-end:
  1. Normal replication of 1000 records — verified by reading the end offset
     of primary.commit-log on the target cluster.
  2. Truncation — pause MM2, produce additional records on the source, run
     kafka-delete-records.sh to lift the low watermark past MM2's last
     position, unpause. Asserts that
     "Source log truncation detected" appears in the task logs and that the
     task transitions to FAILED.
  3. Topic reset — stop MM2, delete and recreate the source topic, produce
     fresh records, restart MM2. Asserts that
     "Source topic reset detected" appears and that the new records arrive
     on the target.
• All scenarios pass deterministically against apache/kafka:4.0.0.
• Existing MirrorMaker 2 integration tests are unaffected (no public API
  changed; the new exception branch is the only execution path that
  behaviour-shifts and it only triggers on a state vanilla MM2 currently
  silently absorbs).

Compatibility

• No public API change.
• No new configuration property.
• Behaviour change is gated on the consumer raising
  OffsetOutOfRangeException, which only happens when the operator sets
  consumer.auto.offset.reset = none. With the default
  (auto.offset.reset = earliest), MirrorSourceTask behaves exactly as
  before.

Committer Checklist (excluded from commit message)

• Verified design and implementation
• Verified test coverage and CI build status
• Verified documentation (including upgrade notes) — N/A, no public surface change